package com.sojson.project.sys.sharding.service.impl;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import javax.annotation.Resource;

import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.sojson.config.datasource.annotation.DataSource;
import com.sojson.config.datasource.enums.EDataSourceType;
import com.sojson.project.sys.sharding.dao.ShardingDao;
import com.sojson.project.sys.sharding.entity.dto.ShardingDto;
import com.sojson.project.sys.sharding.enums.EShardingField;
import com.sojson.project.sys.sharding.enums.EShardingFieldType;
import com.sojson.project.sys.sharding.service.IShardingService;
import com.sojson.util.ExpUtil;
import com.sojson.util.StringUtil;
import com.sojson.util.db.ShardingJdbcUtil;
import com.sojson.util.db.bean.ShardingJdbcVo;

import lombok.extern.slf4j.Slf4j;

/**
 * 分库分表Service实现类
 *
 * @author liu
 * @date 2022-10-24
 */
@Slf4j
@Service
public class ShardingServiceImpl implements IShardingService {

    @Lazy
    @Autowired
    private ShardingServiceImpl shardingService;
    @Resource
    private ShardingDao shardingDao;
    /** 单次查询量 */
    public int searchSize = 2;
    /** 单次处理量 */
    public int disposeSize = 1;
    /** 分表标记 */
    public long shardingTag = System.currentTimeMillis();

    /**
     * 迁移数据
     */
    @Override
    @DataSource(EDataSourceType.master)
    public void execute() {
        log.info("开始执行数据迁移: " + LocalDateTime.now());
        Map<String, ShardingJdbcVo> tables = ShardingJdbcUtil.getTables();
        Set<Entry<String, ShardingJdbcVo>> entrySet = tables.entrySet();
        for (Entry<String, ShardingJdbcVo> table : entrySet) {
            try {
                shardingService.run(table.getValue());
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    @Async("sharding")
    public void run(ShardingJdbcVo table) {
        String tableName = table.getTableName();
        if (shardingDao.tableExists(tableName) == null) {
            return;
        }

        String shardingColumn = table.getShardingColumn();
        long start = System.currentTimeMillis();
        log.info(tableName + "开始执行数据迁移: " + LocalDateTime.now());
        String prefix = "#{data.";
        String suffix = "},";
        String databaseName = shardingDao.getDatabaseName();
        List<String> idName = shardingDao.getIdName(databaseName, tableName);
        String fieldType = shardingDao.getFieldType(databaseName, tableName, shardingColumn);
        String fields = getFields(databaseName, tableName);
        int size = idName.size();
        if (size == 0) {
            ExpUtil.throwEx(tableName + "未设置主键");
        } else if (size != 1) {
            ExpUtil.throwEx(tableName + "有多个主键");
        }

        ShardingDto dto = new ShardingDto();
        dto.setSearchName(shardingColumn);
        dto.setFieldNameId(idName.get(0));
        dto.setTableName(tableName);
        dto.setTableSize(table.getTableNames().size());
        dto.setFieldStrs(fields);
        dto.setShardingTag(shardingTag);
        dto.setSearchSize(searchSize);
        StringBuilder builder = new StringBuilder();
        String[] split = fields.split(",");
        for (String field : split) {
            builder.append(prefix);
            builder.append(field);
            builder.append(suffix);
        }
        dto.setFieldsqls(builder.substring(0, builder.length() - 1));
        EShardingFieldType type = null;
        EShardingField[] values = EShardingField.values();
        for (EShardingField field : values) {
            if (field.getTypeName().equals(fieldType)) {
                type = field.getType();
                break;
            }
        }
        if (type == null) {
            ExpUtil.throwEx("字段类型未设置: " + fieldType);
        }
        if (type == EShardingFieldType.NUM) {
            shardingService.numId(dto, 1);
        } else if (type == EShardingFieldType.NOT_NUM) {
            shardingService.strId(dto);
        } else {
            ExpUtil.throwEx("字段类型不存在");
        }
        log.info(tableName + "数据迁移共用时: " + (System.currentTimeMillis() - start) / 1000 + "秒");
    }

    public void numId(ShardingDto dto, int num) {
        dto.setNum(num);
        String tableName = dto.getTableName();
        List<Map<String, Object>> datas = shardingDao.getDatasByNumId(dto);
        if (datas.size() < 1) {
            if (dto.getTableSize() != num) {
                numId(dto, ++num);
                return;
            } else {
                return;
            }
        }

        dto.setTableNameNew(tableName + StringUtil.SEPARATOR + num);
        List<List<Map<String, Object>>> subLists = Lists.partition(datas, disposeSize);
        for (List<Map<String, Object>> list : subLists) {
            shardingService.save(dto, list);
        }
        shardingService.numId(dto, num);
    }

    public void strId(ShardingDto dto) {
        String tableName = dto.getTableName();
        int size = dto.getTableSize();

        Map<Integer, List<Map<String, Object>>> classify = new HashMap<>();
        for (int i = 0; i < size; i++) {
            classify.put(i + 1, new ArrayList<>());
        }
        List<Map<String, Object>> data = shardingDao.getDatasByStrId(dto);
        if (data.size() < 1) {
            return;
        }
        for (Map<String, Object> map : data) {
            Object object = map.get(dto.getSearchName());
            int hashCode = Math.abs(object.hashCode()) % size + 1;
            List<Map<String, Object>> list = classify.get(hashCode);
            list.add(map);
        }
        Set<Entry<Integer, List<Map<String, Object>>>> entrySet = classify.entrySet();

        dto.setTableName(tableName);
        for (Entry<Integer, List<Map<String, Object>>> entry : entrySet) {
            List<Map<String, Object>> datas = entry.getValue();
            if (datas.size() < 1) {
                continue;
            }

            dto.setTableNameNew(tableName + StringUtil.SEPARATOR + entry.getKey());
            List<List<Map<String, Object>>> subLists = Lists.partition(datas, disposeSize);
            for (List<Map<String, Object>> list : subLists) {
                shardingService.save(dto, list);
            }
        }
        shardingService.strId(dto);
    }

    @Transactional(rollbackFor = Exception.class)
    public void save(ShardingDto dto, List<Map<String, Object>> datas) {
        int save = shardingDao.save(dto, datas);
        if (save != datas.size()) {
            ExpUtil.throwEx("数据迁移失败,插入数量不匹配");
        }

        // 删除历史数据
        List<Object> ids = new ArrayList<>();
        for (Map<String, Object> map : datas) {
            ids.add(map.get(dto.getFieldNameId()));
        }

        shardingDao.remove(dto, ids);
    }

    @Transactional(rollbackFor = Exception.class)
    private String getFields(String databaseName, String tableName) {
        return getFields(databaseName, tableName, null, 0);
    }

    @Transactional(rollbackFor = Exception.class)
    private String getFields(String databaseName, String tableName, String fieldStrs, int num) {
        String fields = shardingDao.getFields(databaseName, tableName, num);
        if (fields == null || fields.length() < 1) {
            return fieldStrs;
        }

        if (fieldStrs == null) {
            return getFields(databaseName, tableName, fields, ++num);
        } else {
            return getFields(databaseName, tableName, fieldStrs + "," + fields, ++num);
        }
    }

}